//package com.lianxi.rocketmqmvc;
//
//import com.alibaba.fastjson.JSON;
//import com.dahe.wang.constant.RocketMQConstant;
//import com.dahe.wang.model.News;
//import com.dahe.wang.util.ResourcesUtil;
//import com.dahe.wang.util.SpringContextHolder;
//import com.google.common.base.Charsets;
//import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
//import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
//import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
//import org.apache.rocketmq.client.exception.MQClientException;
//import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
//import org.apache.rocketmq.common.message.MessageExt;
//import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//import org.springframework.context.annotation.DependsOn;
//import org.springframework.core.task.TaskExecutor;
//
//import java.util.List;
//
///**
// * Description : 消费者工具类
// * Date : 2018/1/23 16:05
// *
// * @author : fy
// */
//@DependsOn("springContextHolder")
//public class ChannelDelConsumer {
//    private static Logger logger = LoggerFactory.getLogger("ChannelDelConsumerUtil");
//
//    private static TaskExecutor taskExecutor = SpringContextHolder.getBean("taskExecutor");
//    /**
//     * 默认name server
//     */
//    private static final String DEFAULT_CONSUMER_NAMESRV_ADDR = ResourcesUtil.getRocketSrvAddr();
//    /**
//     * 单队列并行消费最大跨度, 默认2000
//     */
//    private static final Integer MAX_CONCURRENT_SPAN_SIZE = 2000;
//    /**
//     * 并发消费时，一次消费消息的数量
//     */
//    private static final Integer MAX_MSG_BATCH_SIZE = 100;
//    /**
//     * 每个topic 的最小线程数
//     */
//    private static final Integer DEFAULT_MIN_THREAD_NUM = 10;
//
//    /**
//     * 每个topic的最大线程数
//     */
//    private static final Integer DEFAULT_MAX_THREAD_NUM = 30;
//
//    private static final String SUBSCRIBE_TOPIC = RocketMQConstant.ROCKETMQ_CHANNEL_DEL_TOPIC;
//
//    private static final String CONSUMER_GROUP = RocketMQConstant.ROCKETMQ_CHANNEL_DEL_GROUP;
//
//    private static DefaultMQPushConsumer defaultMQPushConsumer;
//
//    private ChannelDelConsumer() {
//
//    }
//
//    private static DefaultMQPushConsumer getDefaultMQPullConsumer() {
//        if (defaultMQPushConsumer == null) {
//            synchronized (ChannelDelConsumer.class) {
//                if (defaultMQPushConsumer == null) {
//                    defaultMQPushConsumer = new DefaultMQPushConsumer(CONSUMER_GROUP,true);
//                }
//            }
//        }
//        return defaultMQPushConsumer;
//    }
//
//
//
//    /**
//     * 开启消费者
//     */
//    public static void startConsumer() {
//        defaultMQPushConsumer = getDefaultMQPullConsumer();
//        defaultMQPushConsumer.setNamesrvAddr(DEFAULT_CONSUMER_NAMESRV_ADDR);
//        //单队列并行消费最大跨度
//        defaultMQPushConsumer.setConsumeConcurrentlyMaxSpan(MAX_CONCURRENT_SPAN_SIZE);
//        //并发消费时，一次消费消息的数量，默认为1，假如修改为50，此时若有100条消息，那么会创建两个线程，每个线程分配50条消息。
//        defaultMQPushConsumer.setConsumeMessageBatchMaxSize(MAX_MSG_BATCH_SIZE);
//        //从消息队列头开始消费
//        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
//        //消费方式  集群消费
//        defaultMQPushConsumer.setMessageModel(MessageModel.BROADCASTING);
//        //最大消费线程数
//        defaultMQPushConsumer.setConsumeThreadMax(DEFAULT_MAX_THREAD_NUM);
//        //最小消费线程数
//        defaultMQPushConsumer.setConsumeThreadMin(DEFAULT_MIN_THREAD_NUM);
//        //vip通道关闭，开启的话会走netty通道
//        defaultMQPushConsumer.setVipChannelEnabled(false);
//        try {
//            //订阅
//            defaultMQPushConsumer.subscribe(SUBSCRIBE_TOPIC, "*");
//            defaultMQPushConsumer.registerMessageListener((List<MessageExt> list,
//                                                           ConsumeConcurrentlyContext context) -> {
//                list.forEach(messageExt -> {
//                    String msgBody = new String(messageExt.getBody(), Charsets.UTF_8);
//                    News news = JSON.parseObject(msgBody, News.class);
//                });
//                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//            });
//            logger.info(" --------------------- staticConsumer start ---------------------");
//            defaultMQPushConsumer.start();
//        } catch (MQClientException e) {
//            logger.info(e.getMessage());
//        }
//    }
//
//    /**
//     * 关闭消费者
//     */
//    public static void closeConsumer() {
//        defaultMQPushConsumer.shutdown();
//        logger.info(" ----- consumer shutdown ----- ");
//    }
//}
